package cn.com.sparkle.paxos.deamon;

import java.io.IOException;
import java.io.Serializable;
import java.util.LinkedList;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import com.google.protobuf.ByteString;

import cn.com.sparkle.paxos.AccountBook;
import cn.com.sparkle.paxos.Configuration;
import cn.com.sparkle.paxos.ConfigurationException;
import cn.com.sparkle.paxos.addprocess.AddRequestPackage;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.event.InstanceExecuteEvent;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.AddResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.CommandResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SuccessfulRecord;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Value;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ValueType;
import cn.com.sparkle.paxos.model.Command;
import cn.com.sparkle.paxos.stablestorage.ReadSuccessRecordCallback.SuccessfulRecordWrap;
//import cn.com.sparkle.paxos.model.Value;
//import cn.com.sparkle.paxos.net.message.AddResponse;
//import cn.com.sparkle.paxos.net.message.CommandResponse;
import cn.com.sparkle.paxos.userserver.UserHandlerInterface;

public class InstanceExecutor extends Thread {
	private final static Logger logger = Logger
			.getLogger(InstanceExecutor.class);
	private LinkedBlockingQueue<SuccessfulRecordWrap> fifo = new LinkedBlockingQueue<SuccessfulRecordWrap>(
			100);
	private UserHandlerInterface userHandlerInterface;
	private AccountBook abook;
	private Configuration conf;
	private EventsManager eventsManager;

	public InstanceExecutor(UserHandlerInterface userHandlerInterface,
			Configuration conf, EventsManager eventsManager) {
		super();
		this.userHandlerInterface = userHandlerInterface;
		this.conf = conf;
		this.eventsManager = eventsManager;
		this.setName("Paxos-Instance-Executor");
	}

	public void start(AccountBook aBook) {
		this.abook = aBook;
		this.start();
	}

	public void execute(SuccessfulRecordWrap recordWrap) {
		SuccessfulRecord record = recordWrap.getRecord();
		LinkedList<AddRequestPackage> addRequestPackages = recordWrap
				.getAddRequestPackages();

//		logger.debug("execute a record instanceId:"
//				+ recordWrap.getInstanceId());
		if (record.getV().getValues(0).size() == 0) {
			// this is a null command so that it will be not executed
		} else if (record.getV().getType() == ValueType.TYPE_ADMIN) {
			Value v = record.getV();
			Command c = new Command(v.getValues(0).toByteArray());
			try {
				switch (c.getType()) {
				case Command.ADD_FOLLOWER:
					conf.addFollower(c.getAddress());
					break;
				case Command.ADD_SENATOR:
					conf.addSenator(c.getAddress(), recordWrap.getInstanceId());
					break;
				case Command.REMOVE_FOLLOWER:
					conf.removeFollower(c.getAddress());
					break;
				case Command.REMOVE_SENATOR:
					conf.removeSenator(c.getAddress(),
							recordWrap.getInstanceId());
					break;
				default:
					throw new ConfigurationException("This command is invalid!");
				}
				// response success
				CommandResponse response = CommandResponse.newBuilder()
						.setIsSuccessful(true).setError("").build();

				for (AddRequestPackage arp : addRequestPackages) {
					arp.response(response);
				}
			} catch (ConfigurationException e) {
				logger.error("error admin command", e);
				// response fail
				CommandResponse response = CommandResponse.newBuilder()
						.setIsSuccessful(false).setError(e.getMessage())
						.build();
				for (AddRequestPackage arp : addRequestPackages) {
					arp.response(response);
				}
			}
			InstanceExecuteEvent.doEventExecutedEvent(eventsManager, record);
		} else {
//			logger.debug("put execute queue instanceid:"
//					+ recordWrap.getInstanceId());
			try {
				fifo.put(recordWrap);
			} catch (InterruptedException e) {
			}
		}
	}

	public void run() {
		SuccessfulRecordWrap recordWrap;
		while (true) {
			try {
				recordWrap = fifo.take();
				if (recordWrap.getRecord().hasV()
						|| recordWrap.getRecord().getV().getType() != ValueType.TYPE_ADMIN) {// admin
					// command
					// has be
					// executed

					userHandlerInterface.onLoged(recordWrap,
							recordWrap.getInstanceId(), recordWrap.getRecord()
									.getV().getValuesList(), abook);

					/*
					 * byte[][] result = new
					 * byte[recordWrap.getRecord().getV().getValuesCount()][];
					 * for(int i = 0 ; i < result.length ; ++i){ result[i] =
					 * execute.execute(recordWrap.getRecord().getInstanceId(),
					 * recordWrap
					 * .getRecord().getV().getValues(i).toByteArray()); }
					 * 
					 * if (recordWrap.getAddRequestPackages()!= null) { // send
					 * add response LinkedList<AddRequestPackage>
					 * addRequestPackages = recordWrap.getAddRequestPackages();
					 * 
					 * AddResponse.Builder addResponseBuilder =
					 * AddResponse.newBuilder(); for (AddRequestPackage arp :
					 * addRequestPackages) { for (ByteString bb : arp.valueList)
					 * { // ByteString bs =
					 * ByteString.copyFrom(result[resultIndex]);
					 * addResponseBuilder.setResult(bb); AddResponse response =
					 * addResponseBuilder.build();
					 * userHandlerInterface.onLoged(arp.getSession(), ,
					 * packageId, value, aBook); // arp.response(response);
					 * ++responseCount; } } }else{
					 * 
					 * }
					 */

					// logger.debug("response instanceId:"
					// + recordWrap.getRecord().getInstanceId());
					InstanceExecuteEvent.doEventExecutedEvent(eventsManager,
							recordWrap.getRecord());
				}
				// abook.finishCurInstance();
			} catch (InterruptedException e) {
				logger.error("fatal error", e);
			} catch (IOException e) {
				logger.error("fatal error", e);
			}
		}
	}
}
